package com.ftg.learn.execl.service;

import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.context.AnalysisContext;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ftg.learn.execl.domain.StudentDomain;
import com.ftg.learn.execl.util.ExeclUtil;
import com.google.common.util.concurrent.RateLimiter;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
import com.google.gson.reflect.TypeToken;
import io.micrometer.core.instrument.Metrics;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.retry.RetryException;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.data.redis.connection.stream.StreamRecords;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 安装 rabbitmq
 * docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:3-management
 * 启用管理页面
 * rabbitmq-plugins enable rabbitmq_management
 * 优化方案 以及细节
 *
 * @author 布衣
 */
@Slf4j
@Service
public class ExcelServiceOpt {

    private ExecutorService excelExecutor;
    @Value("${excel.thread.core:8}")
    private int corePoolSize;
    @Value("${excel.thread.max:16}")
    private int maxPoolSize;
    //最大重试数量
    private final int MAX_RETRY = 5;
    // 初始退避时间：1秒
    private static final long INITIAL_BACKOFF_MS = 1000;
    // 退避乘数：2倍
    private static final double BACKOFF_MULTIPLIER = 2.0;

    //锁... 本地锁
    @Resource
    private RedisTemplate<String, String> redisLockTemplate;

    @Resource
    private ObjectMapper objectMapper;
    @Resource
    private JdbcTemplate jdbcTemplate;

    // 优化点3：Redis Stream替代Hash结构
    private static final String STREAM_KEY = "excel:stream";
    private static final String GROUP_NAME = "group1";
    private static final String CONSUMER_NAME = "consumer1";

    // 优化点4：限流器
    private RateLimiter rateLimiter;
    @Value("${excel.rate.limit:5000}")
    private double rateLimit;

    // 优化点5：监控指标
    private AtomicLong processedRows = new AtomicLong(0);
    private AtomicLong errorCount = new AtomicLong(0);

    // 优化点6：批次控制
    @Value("${excel.batch.size:1000}")
    private int batchSize;


    /**
     * 初始化方法，主要用于设置线程池、检查并初始化Stream、注册消费者以及设置速率限制和监控指标
     * 该方法在对象创建时自动调用，确保所有必要的资源和配置都被正确设置
     */
    @PostConstruct
    public void init() {
        // 创建可扩容线程池
        excelExecutor = new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                new ThreadFactoryBuilder().setNameFormat("excel-process-%d").build(),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        try {
            // 1. 检查Stream是否存在
            if (!isStreamExists()) {
                log.info("Stream {} does not exist, creating...", STREAM_KEY);
                createStreamWithGroup();
            } else {
                log.info("Stream {} already exists", STREAM_KEY);
                createGroupIfNotExists();
            }

            // 2. 注册消费者
            registerConsumer();
        } catch (Exception e) {
            log.error("Stream initialization failed", e);
            throw new RuntimeException("Stream initialization failed", e);
        }

        // 初始化速率限制器
        rateLimiter = RateLimiter.create(rateLimit);

        // 初始化监控指标
        Metrics.gauge("excel.processed.rows", processedRows);
        Metrics.gauge("excel.error.count", errorCount);
    }

    /**
     * 判断Redis中的Stream是否存在
     *
     * 该方法尝试通过调用redisLockTemplate的opsForStream().info()方法来检查指定的Stream键是否存在
     * 如果Stream键存在，则返回true；如果不存在，则返回false
     * 此方法主要用于在执行Stream相关操作前，确认Stream是否已经创建
     *
     * @return boolean 表示Stream是否存在的布尔值，存在返回true，否则返回false
     */
    private boolean isStreamExists() {
        try {
            // 尝试获取Stream的信息，如果Stream存在，则不抛出异常
            redisLockTemplate.opsForStream().info(STREAM_KEY);
            return true;
        } catch (Exception e) {
            // 捕获异常，检查异常消息是否包含"no such key"
            // 如果包含，说明Stream不存在，返回false
            if (e.getMessage().contains("no such key")) {
                return false;
            }
            // 如果异常消息不包含"no such key"，则重新抛出异常
            // 这样做是为了让调用者知道在检查Stream存在性时发生了其他错误
            throw e;
        }
    }

    /**
     * 创建带有消费者组的Stream
     * 此方法首先在Stream中添加一个初始记录，然后创建一个消费者组，并最后删除这个初始记录
     */
    private void createStreamWithGroup() {
        // 使用特殊ID "0-0" 创建空Stream
        RecordId initialId = redisLockTemplate.opsForStream()
                .add(StreamRecords.newRecord()
                        .ofObject("initial_data")
                        .withStreamKey(STREAM_KEY));

        // 创建消费者组（从起始位置消费）
        redisLockTemplate.opsForStream().createGroup(STREAM_KEY, ReadOffset.from("0-0"), GROUP_NAME);
        log.info("Created stream {} with initial group {}", STREAM_KEY, GROUP_NAME);

        // 删除初始测试数据
        redisLockTemplate.opsForStream().delete(STREAM_KEY, initialId);
    }

    /**
     * 如果消费者组不存在，则创建消费者组
     * 此方法首先检查指定的流中是否存在具有指定名称的消费者组
     * 如果消费者组不存在，它将创建一个新的消费者组，并记录创建信息
     * 如果消费者组已经存在，则记录该信息
     * 在检查或创建消费者组过程中遇到的任何异常都将被捕获并记录，然后重新抛出
     */
    private void createGroupIfNotExists() {
        try {
            // 获取流中的消费者组信息
            StreamInfo.XInfoGroups groups = redisLockTemplate.opsForStream().groups(STREAM_KEY);

            // 检查是否存在具有指定名称的消费者组
            boolean groupExists = groups.stream()
                    .anyMatch(g -> GROUP_NAME.equals(g.groupName()));

            // 如果消费者组不存在，则创建它
            if (!groupExists) {
                redisLockTemplate.opsForStream().createGroup(STREAM_KEY, GROUP_NAME);
                log.info("Created consumer group {} on existing stream {}", GROUP_NAME, STREAM_KEY);
            } else {
                log.info("Consumer group {} already exists", GROUP_NAME);
            }
        } catch (Exception e) {
            // 记录在检查或创建消费者组过程中遇到的异常
            log.error("Failed to create group", e);
            // 重新抛出异常，以便可以在调用此方法的代码中处理
            throw e;
        }
    }

    /**
     * 注册消费者到Redis流
     *
     * 此方法检查指定的消费者是否已存在于Redis流的消费者组中如果不存在，则通过执行PING命令隐式注册该消费者
     * 隐式注册是因为Redis没有直接的命令来注册消费者，但执行某些命令（如PING）会自动在服务器上注册该消费者
     */
    private void registerConsumer() {
        try {
            // 获取指定流和消费者组的消费者信息
            StreamInfo.XInfoConsumers consumers = redisLockTemplate.opsForStream()
                    .consumers(STREAM_KEY, GROUP_NAME);

            // 检查消费者是否已经存在
            boolean consumerExists = consumers.stream()
                    .anyMatch(c -> CONSUMER_NAME.equals(c.consumerName()));

            // 如果消费者不存在，则执行PING命令以隐式注册消费者
            if (!consumerExists) {
                redisLockTemplate.opsForStream()
                        .pending(STREAM_KEY, GROUP_NAME);
                log.info("Registered new consumer: {}", CONSUMER_NAME);
            }
        } catch (Exception e) {
            // 捕获并记录消费者注册检查过程中的异常
            log.warn("Consumer registration check failed", e);
        }
    }

    /**
     * 读取所有工作表
     * 本方法通过分布式锁确保并发情况下文件处理的唯一性，并采用EasyExcel库进行Excel文件的读取
     * 使用自定义Excel监听器实现分片处理和流式写入，以提高处理效率和减少内存消耗
     *
     * @param filePath Excel文件路径
     * @return 处理结果字符串，成功时返回"success"
     * @throws RuntimeException 当Excel处理失败时抛出运行时异常
     */
    public String readAllSheets(String filePath) {
        // 根据文件路径生成唯一的文件密钥
        final String fileKey = DigestUtils.sha256Hex(filePath);
        // 定义锁的键值，用于获取分布式锁
        final String lockKey = "lock:excel:" + fileKey;

        // 使用分布式锁执行Excel处理操作，确保并发情况下文件处理的安全性
        return executeWithDistributedLock(lockKey, () -> {
            try {
                // 创建一个匿名内部类作为EasyExcel的监听器，实现对Excel数据的自定义处理
                ExeclUtil listener = new ExeclUtil() {
                    // 定义一个缓冲区用于暂存读取到的学生数据
                    private final List<StudentDomain> buffer = new ArrayList<>(batchSize);
                    // 创建Gson实例用于JSON序列化
                    private final Gson gson = new Gson();

                    /**
                     * 处理每个学生数据
                     * 在数据被添加到缓冲区之前，进行数据校验如果数据校验失败，则增加错误计数并忽略当前数据
                     * 当缓冲区大小达到预设的批次大小时，调用processBatch方法处理当前批次的数据，并清空缓冲区
                     *
                     * @param data 学生数据对象
                     * @param context 上下文对象，提供与分析相关的上下文信息
                     */
                    @Override
                    public void invoke(StudentDomain data, AnalysisContext context) {
                        if (!validateData(data)) {
                            errorCount.incrementAndGet();
                            return;
                        }

                        buffer.add(data);
                        if (buffer.size() >= batchSize) {
                            processBatch(buffer, fileKey);
                            buffer.clear();
                        }
                        processedRows.incrementAndGet();
                    }

                    /**
                     * 当所有数据分析完毕后调用此方法
                     * 检查缓冲区中是否还有剩余的数据，如果有，则进行处理
                     *
                     * @param context 上下文对象，提供与分析相关的上下文信息
                     */
                    @Override
                    public void doAfterAllAnalysed(AnalysisContext context) {
                        if (!buffer.isEmpty()) {
                            processBatch(buffer, fileKey);
                        }
                    }

                    /**
                     * 处理单个批次的数据
                     * 本方法首先进行限流控制，然后异步执行数据处理和写入操作
                     *
                     * @param batch 当前批次的学生数据列表
                     * @param fileKey 文件密钥，用于标识当前处理的Excel文件
                     */
                    private void processBatch(List<StudentDomain> batch, String fileKey) {
                        // 限流控制
                        rateLimiter.acquire(batch.size());

                        excelExecutor.execute(() -> {
                            String json = gson.toJson(batch);
                            writeToStream(json, fileKey);
                        });
                    }
                };

                // 使用EasyExcel读取Excel文件，指定监听器处理数据
                EasyExcel.read(filePath, StudentDomain.class, listener)
                        .sheet()
                        .headRowNumber(5)
                        .doRead();

                return "success";
            } catch (Exception e) {
                // 日志记录Excel处理失败的信息，并抛出运行时异常
                log.error("Excel processing failed: {}", filePath, e);
                errorCount.incrementAndGet();
                throw new RuntimeException("Excel processing failed", e);
            }
        });
    }

    /**
     * 使用分布式锁执行任务
     * 此方法旨在确保任务在分布式环境中互斥执行，通过Redis实现锁机制
     *
     * @param lockKey 锁的键，用于标识资源
     * @param task 需要执行的任务，实现Callable接口
     * @return 任务执行结果
     * @throws RuntimeException 如果未能获取锁或执行任务时出现异常
     */
    private String executeWithDistributedLock(String lockKey, Callable<String> task) {
        // 生成唯一的请求ID，用于标识此次锁操作
        String requestId = UUID.randomUUID().toString();
        try {
            // 尝试获取分布式锁，如果锁不存在，则设置锁并返回true；否则返回false
            Boolean locked = redisLockTemplate.opsForValue()
                    .setIfAbsent(lockKey, requestId, Duration.ofMinutes(5));
            if (Boolean.TRUE.equals(locked)) {
                // 获取锁成功，执行任务
                return task.call();
            }
            // 获取锁失败，抛出异常
            throw new RuntimeException("Acquire lock failed");
        } catch (Exception e) {
            // 处理任务执行过程中出现的异常
            throw new RuntimeException("Lock operation error", e);
        } finally {
            // 检查当前锁是否由此次请求持有，如果是，则释放锁
            if (requestId.equals(redisLockTemplate.opsForValue().get(lockKey))) {
                redisLockTemplate.delete(lockKey);
            }
        }
    }

    /**
     * 将数据写入到指定的流中
     * 此方法使用了重试机制，以处理可能的查询超时异常
     * 选择的异常类型、重试次数和重试间隔是为了确保数据写入的可靠性
     *
     * @param json 要写入的JSON格式的数据
     * @param fileKey 数据的唯一标识符，用于流中的记录
     */
    @Retryable(value = {QueryTimeoutException.class},
            maxAttempts = 3,
            backoff = @Backoff(delay = 100, maxDelay = 1000))
    private void writeToStream(String json, String fileKey) {
        // 创建一个流记录，包含要写入的数据
        MapRecord<String, String, String> record = StreamRecords.newRecord()
                .ofMap(Collections.singletonMap("data", json))
                .withStreamKey(STREAM_KEY)
                .withId(RecordId.autoGenerate());

        // 使用Redis模板将记录添加到流中
        redisLockTemplate.opsForStream()
                .add(record);
    }

    // 定义一个定时任务方法，每5秒执行一次
    @Scheduled(fixedDelay = 5000)
    public void scheduleConsume() {
        // 从Redis流中读取消息记录
        // 这里使用了Redis的流操作模板redisLockTemplate，配置了消费者组和消费者名称
        // 读取消息的偏移量设置为上次消费的位置，以实现持续消费
        List<MapRecord<String, Object, Object>> records = redisLockTemplate.opsForStream()
                .read(Consumer.from(GROUP_NAME, CONSUMER_NAME),
                        StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()));

        // 检查是否有消息记录读取到
        if (!records.isEmpty()) {
            // 处理读取到的消息记录
            processRecords(records);
            // 确认处理的消息，以确保消息被正确消费
            acknowledgeMessages(records);
        }
    }

    /**
     * 确认消息消费完成
     *
     * 此方法用于在消费者处理完消息后，进行确认操作，以确保消息不会被重复消费
     *
     * @param records 一个包含多个消息记录的列表，每个记录都是一个MapRecord，包含消息的键、值和属性
     */
    private void acknowledgeMessages(List<MapRecord<String, Object, Object>> records) {
        // 打印消息确认消费的信息，这里应该有更具体的确认逻辑
        System.out.println("我是消费了...你看我走没走"+records);
    }

    /**
     * 处理记录列表
     * 该方法主要负责解析和处理接收到的数据记录，过滤出有效数据，并将其插入到数据库中
     *
     * @param records 记录列表，每个记录是一个映射，包含数据的键值对
     */
    private void processRecords(List<MapRecord<String, Object, Object>> records) {
        // 使用Gson进行JSON解析
        Gson gson = new Gson();

        // 解析并过滤有效记录
        List<StudentDomain> validRecords = records.stream()
                .flatMap(record -> {
                    try {
                        // 判断 data 是否为 JSON 数组
                        String jsonData = record.getValue().get("data").toString();
                        JsonElement jsonElement = JsonParser.parseString(jsonData);

                        // 根据数据类型解析JSON
                        if (jsonElement.isJsonArray()) {
                            // 如果是数组，解析为 List<StudentDomain>
                            List<StudentDomain> studentList = gson.fromJson(jsonElement, new TypeToken<List<StudentDomain>>() {}.getType());
                            return studentList.stream();
                        } else if (jsonElement.isJsonObject()) {
                            // 如果是对象，直接解析为 StudentDomain
                            StudentDomain student = gson.fromJson(jsonElement, StudentDomain.class);
                            return Stream.of(student);
                        } else {
                            // 无效数据，返回空流
                            return Stream.empty();
                        }
                    } catch (Exception e) {
                        // 解析失败，记录日志并跳过
                        System.err.println("解析失败: " + e.getMessage());
                        return Stream.empty();
                    }
                })
                // 过滤掉空值和无效数据
                .filter(Objects::nonNull)
                .filter(this::validateData)
                .collect(Collectors.toList());

        // 如果没有有效记录，记录日志并返回
        if (validRecords.isEmpty()) {
            log.warn("本批次无有效数据 records:{}", records.size());
            return;
        }

        // 使用分布式事务写入有效记录到数据库
        try {
            // 带重试机制的批量写入
            executeWithRetry(() -> {
                batchInsert("student", validRecords);
                return null;
            });
        } catch (Exception e) {
            // 处理批量插入失败的情况
            handleFailedBatch(validRecords, e);
        }
    }

    /**
     * 处理批次写入失败的情况
     * 当批次写入操作失败时，此方法将被调用以记录错误信息，并将失败的记录写入死信队列
     *
     * @param failedBatch 失败的批次，包含一批学生域对象
     * @param e 引发批次写入失败的异常
     */
    private void handleFailedBatch(List<StudentDomain> failedBatch, Exception e) {
        // 记录批次写入失败的日志信息
        log.error("批次写入失败 数量:{} 原因:{}", failedBatch.size(), e.getMessage());

        // 将失败记录写入死信队列
        List<ObjectRecord<String, String>> deadLetters = failedBatch.stream()
                .map(record -> {
                    try {
                        // 序列化失败的记录，并创建死信队列记录
                        return StreamRecords.newRecord()
                                .ofObject(objectMapper.writeValueAsString(record))
                                .withStreamKey("dead_letter_stream");
                    } catch (JsonProcessingException ex) {
                        // 如果序列化失败，记录错误日志
                        log.error("死信序列化失败", ex);
                        return null;
                    }
                })
                .filter(Objects::nonNull) // 过滤掉序列化失败的记录
                .collect(Collectors.toList());

        // 如果存在有效的死信记录，则将其添加到Redis的死信队列中，并更新监控指标
        if (!deadLetters.isEmpty()) {
            deadLetters.forEach(record -> {
                redisLockTemplate.opsForStream().add(record);
            });
            Metrics.counter("dead_letter_count").increment(deadLetters.size());
        }
    }

    /**
     * 执行带有重试机制的任务
     * 当任务执行失败时，根据指数退避策略进行重试，直到达到最大重试次数
     *
     * @param task 要执行的任务，一个 Callable 接口的实现
     * @param <T> 任务返回值的类型
     * @return 任务的执行结果
     * @throws Exception 如果任务执行失败，并且重试次数达到最大值，则抛出异常
     */
    private <T> T executeWithRetry(Callable<T> task) throws Exception {
        // 初始化重试计数
        int retryCount = 0;
        // 保存最后一次异常，以便于重试次数耗尽时抛出
        Exception lastException = null;

        // 循环尝试执行任务，直到成功或达到最大重试次数
        while (retryCount <= MAX_RETRY) {
            try {
                // 尝试执行任务并返回结果
                return task.call();
            } catch (Exception e) {
                // 捕获异常，记录最后一次异常信息
                lastException = e;
                // 记录日志，警告操作失败，并指出重试次数和失败原因
                log.warn("操作失败 重试次数:{}/{} 原因:{}",
                        retryCount, MAX_RETRY, e.getMessage());
                // 根据重试次数计算休眠时间，并使当前线程休眠
                Thread.sleep(calculateBackoff(retryCount));
                // 增加重试计数
                retryCount++;
            }
        }
        // 如果达到这里，说明重试次数耗尽，抛出自定义异常
        throw new RetryException("重试次数耗尽", lastException);
    }

    /**
     * 计算重试操作的退避时间
     * 该方法用于根据重试次数计算出适当的退避（重试间隔）时间，以避免因频繁请求导致的问题
     * 退避时间通过将初始退避时间与退避乘数的重试次数次幂相乘得到，确保每次重试后退避时间呈指数增长
     *
     * @param retryCount 重试次数，用于计算退避时间
     * @return 退避时间，单位为毫秒
     */
    private long calculateBackoff(int retryCount) {
        // 计算退避时间，使用指数退避策略以分散重试请求，减少对系统的冲击
        return (long) (INITIAL_BACKOFF_MS * Math.pow(BACKOFF_MULTIPLIER, retryCount));
    }

    /**
     * 批量插入数据到指定表中
     *
     * @param tableName 要插入数据的表名
     * @param batch     待插入的学生域名对象列表
     */
    private void batchInsert(String tableName, List<StudentDomain> batch) {
        // 格式化SQL插入语句
        String sql = String.format("insert INTO %s (id, name, gender, classname) VALUES (?, ?, ?, ?)", tableName);

        // 使用JdbcTemplate的batchUpdate方法执行批量更新
        jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
            /**
             * 设置预处理语句的值
             *
             * @param ps 预处理语句对象
             * @param i  当前遍历的索引
             * @throws SQLException 如果设置参数时发生错误
             */
            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                // 获取当前索引对应的学生域名对象
                StudentDomain record = batch.get(i);
                // 为预处理语句设置参数值
                ps.setString(1, record.getId());
                ps.setString(2, record.getName());
                ps.setString(3, record.getGender());
                ps.setString(4, record.getClassName());
            }

            /**
             * 获取批量更新的大小
             *
             * @return 批量更新的大小，即待插入的学生域名对象列表的大小
             */
            @Override
            public int getBatchSize() {
                return batch.size();
            }
        });
    }

    /**
     * 验证学生信息的有效性
     * 此方法检查学生域对象中的必要信息是否完整，以确保后续操作可以顺利进行
     *
     * @param data 学生域对象，包含学生相关信息
     * @return boolean 表示学生信息是否有效的布尔值如果学生姓名非空且ID不为null，则返回true；否则返回false
     */
    private boolean validateData(StudentDomain data) {
        // 检查学生姓名是否非空
        // 检查学生ID是否不为null
        return !StringUtils.isEmpty(data.getName())
                && data.getId() != null;
    }

    /**
     * 定期记录处理状态的指标信息
     * 该方法使用了@Scheduled注解，设置fixedRate为60_000毫秒（即每分钟执行一次）
     * 其目的是定期输出处理的成功行数和错误数量，以便于监控和分析系统性能
     */
    @Scheduled(fixedRate = 60_000)
    public void logMetrics() {
        // 输出处理状态信息，包括成功处理的行数和错误数量
        log.info("Processing status - Success: {}, Errors: {}",
                processedRows.get(), errorCount.get());
    }
}
